package rabbitmq

import (
	"errors"
	"fmt"
	"log"
	"sync"

	"github.com/streadway/amqp"
)

// 单例实例
var (
	instance *RabbitClient
	once     sync.Once
	maxRetry = 3 // 最大重试次数
)

// Config RabbitMQ 配置
type Config struct {
	Username    string
	Password    string
	Host        string
	Port        string
	VHost       string // 虚拟主机，默认为 "/"
	DeadQueue   *DeadQueue
	QueueConfig *QueueConfig
}

type DeadQueue struct {
	DlxExchange string
	Direct      string
	QueueName   string
	RoutingKey  string
}

type QueueConfig struct {
	Exchange  string
	QueueName string
	TTL       int64
}

// RabbitClient 单例客户端
type RabbitClient struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	mutex   sync.Mutex
	config  Config
}

// GetInstance 获取单例实例
func GetInstance(cfg Config) *RabbitClient {
	once.Do(func() {
		instance = &RabbitClient{
			config: cfg,
		}
		// 初始化连接
		if err := instance.initConnection(); err != nil {
			log.Fatalf("初始化 RabbitMQ 连接失败: %v", err)
		}
	})
	return instance
}

// 初始化连接
func (c *RabbitClient) initConnection() error {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	// 构建连接地址
	vhost := c.config.VHost
	if vhost == "" {
		vhost = "/"
	}
	addr := fmt.Sprintf(
		"amqp://%s:%s@%s:%s%s",
		c.config.Username,
		c.config.Password,
		c.config.Host,
		c.config.Port,
		vhost,
	)

	var err error
	c.conn, err = amqp.Dial(addr)
	if err != nil {
		return fmt.Errorf("连接失败: %w", err)
	}

	// 创建通道
	c.channel, err = c.conn.Channel()
	if err != nil {
		c.conn.Close()
		return fmt.Errorf("创建通道失败: %w", err)
	}

	if c.config.QueueConfig.TTL == 0 {
		c.config.QueueConfig.TTL = 3000 // 过期时间，默认3s
	}

	if c.config.DeadQueue.DlxExchange == "" {
		c.config.DeadQueue.DlxExchange = "dlx_exchange"
	}
	if c.config.DeadQueue.Direct == "" {
		c.config.DeadQueue.Direct = "direct"
	}
	if c.config.DeadQueue.QueueName == "" {
		c.config.DeadQueue.QueueName = "dead_letter_queue"
	}
	if c.config.DeadQueue.RoutingKey == "" {
		c.config.DeadQueue.RoutingKey = "dlq_routing_key"
	}
	c.initDLQ(c.channel)

	return nil
}

// 确保连接有效
func (c *RabbitClient) ensureConnection() error {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.conn == nil || c.conn.IsClosed() {
		log.Println("连接已关闭，尝试重连...")
		if err := c.initConnection(); err != nil {
			return err
		}
		log.Println("重连成功")
	}
	return nil
}

func (c *RabbitClient) initDLQ(ch *amqp.Channel) {
	// 1. 声明死信交换机（DLX）
	err := ch.ExchangeDeclare(
		c.config.DeadQueue.DlxExchange, // 交换机名称
		c.config.DeadQueue.Direct,      // 类型（直接路由）
		true,                           // 持久化
		false,                          // 不自动删除
		false,                          // 非内部
		false,                          // 非阻塞
		nil,
	)
	if err != nil {
		log.Fatalf("声明死信交换机失败: %v", err)
	}
	// 2. 声明死信队列（DLQ）
	dlq, err := ch.QueueDeclare(
		c.config.DeadQueue.QueueName, // 死信队列名称
		true,                         // 持久化
		false,                        // 不自动删除
		false,                        // 非排他
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("声明死信队列失败: %v", err)
	}
	// 3. 绑定死信队列到死信交换机
	err = ch.QueueBind(
		dlq.Name,
		c.config.DeadQueue.RoutingKey, // 死信路由键
		c.config.DeadQueue.DlxExchange,
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("绑定死信队列到死信交换机失败: %v", err)
	}
}

// DeclareQueue 声明队列
func (c *RabbitClient) DeclareQueue(queueName string, durable bool) error {
	if err := c.ensureConnection(); err != nil {
		return err
	}
	_, err := c.channel.QueueDeclare(
		queueName, // 队列名称
		durable,   // 持久化
		false,     // 自动删除
		false,     // 排他性
		false,     // 非阻塞
		amqp.Table{
			// 关联死信交换机
			"x-dead-letter-exchange":    c.config.DeadQueue.DlxExchange,
			"x-dead-letter-routing-key": c.config.DeadQueue.RoutingKey,
			"x-message-ttl":             c.config.QueueConfig.TTL, // 消息过期时间（5秒，重试间隔）
		},
	)
	return err
}

// SendMessage 发送消息
func (c *RabbitClient) SendMessage(queueName string, body []byte) error {
	if err := c.ensureConnection(); err != nil {
		return err
	}

	return c.channel.Publish(
		c.config.QueueConfig.Exchange, // 交换机
		queueName,                     // 路由键
		false,                         // mandatory
		false,                         // immediate
		amqp.Publishing{
			ContentType:  "text/plain",
			Body:         body,
			DeliveryMode: amqp.Persistent, // 持久化消息
		},
	)
}

// Consumer 消费者结构体
type Consumer struct {
	QueueName    string
	WorkerCount  int        // 消费者数量
	AutoAck      bool       // 是否自动确认
	HandleFunc   HandleFunc // 消息处理函数
	ErrorHandler ErrorFunc  // 错误处理函数
	messages     <-chan amqp.Delivery
	wg           sync.WaitGroup
	stopChan     chan struct{}
}

// HandleFunc 消息处理函数类型
type HandleFunc func(body []byte) error

// ErrorFunc 错误处理函数类型
type ErrorFunc func(error)

// NewConsumer 创建新消费者
func (c *RabbitClient) NewConsumer(queueName string, workerCount int, autoAck bool) (*Consumer, error) {
	if workerCount <= 0 {
		return nil, errors.New("消费者数量必须大于0")
	}

	// 确保队列存在
	if err := c.DeclareQueue(queueName, true); err != nil {
		return nil, err
	}

	// 注册消费者
	msgs, err := c.channel.Consume(
		queueName, // 队列名称
		"",        // 消费者标签
		autoAck,   // 自动确认
		false,     // 排他性
		false,     // no-local
		false,     // no-wait
		nil,       // args
	)
	if err != nil {
		return nil, err
	}

	return &Consumer{
		QueueName:   queueName,
		WorkerCount: workerCount,
		AutoAck:     autoAck,
		messages:    msgs,
		stopChan:    make(chan struct{}),
	}, nil
}

// Start 启动消费者
func (c *Consumer) Start() {
	// 启动指定数量的工作协程
	for i := 0; i < c.WorkerCount; i++ {
		c.wg.Add(1)
		go func(workerID int) {
			defer c.wg.Done()
			c.processMessages(workerID)
		}(i)
	}
	log.Printf("已启动 %d 个消费者处理队列: %s", c.WorkerCount, c.QueueName)
}

// 处理消息
func (c *Consumer) processMessages(workerID int) {
	for {
		select {
		case d, ok := <-c.messages:
			if !ok {
				log.Printf("消费者 %d 消息通道已关闭", workerID)
				return
			}

			// 调用消息处理函数
			if c.HandleFunc != nil {
				if err := c.HandleFunc(d.Body); err != nil {
					log.Printf("消费者 %d 处理消息失败: %v", workerID, err)
					if c.ErrorHandler != nil {
						c.ErrorHandler(err)
					}
					// 如果不是自动确认，处理失败则拒绝消息
					if !c.AutoAck {
						d.Nack(false, true) // 拒绝消息并重新入队
					}
					continue
				}
			}

			// 如果不是自动确认，处理成功则手动确认
			if !c.AutoAck {
				if err := d.Ack(false); err != nil {
					log.Printf("消费者 %d 确认消息失败: %v", workerID, err)
				}
			}

		case <-c.stopChan:
			log.Printf("消费者 %d 收到停止信号", workerID)
			return
		}
	}
}

// Stop 停止消费者
func (c *Consumer) Stop() {
	close(c.stopChan)
	c.wg.Wait()
	log.Printf("所有消费者已停止 (队列: %s)", c.QueueName)
}

// Close 关闭连接
func (c *RabbitClient) Close() error {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	if c.channel != nil {
		c.channel.Close()
	}
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

// 从消息头获取重试次数
func getRetryCount(headers amqp.Table) int {
	if headers == nil {
		return 0
	}

	// x-death 是一个数组，记录消息死亡历史
	xDeath, ok := headers["x-death"].([]interface{})
	if !ok || len(xDeath) == 0 {
		return 0
	}

	// 取最新的一次死亡记录
	lastDeath, ok := xDeath[0].(amqp.Table)
	if !ok {
		return 0
	}

	// count 字段记录重试次数
	count, ok := lastDeath["count"].(int64)
	if !ok {
		return 0
	}
	return int(count)
}
